package controller

import (
	kc "common/kafkacluster/consumer"
	log "common/log4go"
	"iot-bridge/iot_cmd_up/ping/service"
	"time"
)

type Factory struct {
	closed  bool
	works   map[int]*Work
	monitor chan int
	conf    kc.Config
}

func NewFactory(c kc.Config) (factory *Factory) {
	factory = &Factory{}
	factory.conf = c
	return
}

func (this *Factory) newWork(seqNo int, monitor chan int) (work *Work, err error) {

	work = &Work{}
	consumer, err := kc.NewConsumer(this.conf)
	if err != nil {
		log.Error("kc.NewConsumer(%v) error(%v)", this.conf, err)
		return
	}
	work.consumer = consumer

	work.dataServ = service.NewPingUpServ()

	work.seqNo = seqNo
	work.monitor = monitor
	work.finishOneTask = true
	return
}

func (this *Factory) CreateWorks(count int) (err error) {
	monitor := make(chan int, count)
	this.monitor = monitor
	var work *Work
	for i := 1; i <= count; i++ {
		work, err = this.newWork(i, this.monitor)
		if err != nil {
			log.Error("newWork(c,%d,this.monitor) error(%v)", i, err)
			return
		}

		if this.works == nil {
			tmp := make(map[int]*Work)
			this.works = tmp
		}
		this.works[i] = work
	}
	return
}

func (this *Factory) Begin() {
	if len(this.works) == 0 {
		log.Error("no works do task")
		return
	}
	for _, v := range this.works {
		go v.StartWork(&this.closed)
	}

	go func() {
		for {
			seqNo, ok := <-this.monitor
			if !ok {
				break
			}
			//restart
			for {
				log.Info("worker(%d) is waiting for resource for restart", seqNo)
				time.Sleep(30 * time.Second)
				work, err := this.newWork(seqNo, this.monitor)
				if err != nil {
					log.Error("newWork(%d,this.monitor) error(%v)", seqNo, err)
					continue
				}

				if this.works == nil {
					tmp := make(map[int]*Work)
					this.works = tmp
				}
				this.works[seqNo] = work
				log.Info("workder(%d) restart ok", work.seqNo)
				go work.StartWork(&this.closed)
				break
			}
		}
	}()
	return
}

func (this *Factory) Close() {
	this.closed = true //停工通知
	//切断每个工人的用料来源,即 每个工人受消息的通道会关闭，但是得保证每个工人完成最后的提交
	log.Info("factory will close")
	for _, v := range this.works {
		for {
			if v.finishOneTask {
				v.Stop()
				break
			}
			log.Info("wait for worker(%d) finished his work", v.seqNo)
			time.Sleep(1 * time.Second)
		}
	}

	log.Info("factory all worker stop ok")
	close(this.monitor)
	log.Info("factory close monitor ok")
	time.Sleep(1 * time.Second)
	log.Info("factory close  ok")
}
